package com.doom.a3rdlibdemo;

import com.doom.a3rdlibdemo.decorator.BasePizza;

public class ObservableMap<T, R> extends AbstractObservableWithUpstream<T, R> {

    final Function<T, R> function;

    public ObservableMap(ObservableSource<T> source, Function<T, R> function) {
        super(source);
        this.function = function;
    }

    @Override
    protected void subscribeActual(Observer<T> observer) {
        source.subscribe(new MapObserver<>(observer, function));
    }

    static final class MapObserver<T, R> extends BasicFuseableObserver<T, R> {

        final Function<T, R> mapper;

        public MapObserver(Observer<T> actual, Function<T, R> mapper) {
            super(actual);
            this.mapper = mapper;
        }

        @Override
        public void onNext(T t) {
            R apply = mapper.apply(t);
            actual.onNext((T) apply);
        }
    }
}
